# _*_ coding:utf-8 _*_
# @Time  : 2022.08.11
# @Author: zizlee
import datetime
import json
import time
import pandas as pd
import requests
import pathlib
from urllib3 import disable_warnings
from itertools import groupby
from zizlee_position import get_dominant_price_position, get_contract_price_position

disable_warnings()

USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) " \
             "Chrome/78.0.3904.108 Safari/537.36"


def split_number_en(ustring):
    return [''.join(list(g)) for k, g in groupby(ustring, key=lambda x: x.isdigit())]


def get_variety_list(current_date):
    # VARIETY_LIST = ["IF"]  # 20100416-20130905
    # VARIETY_LIST = ["IF", "TF"]  # 20130906-20150319
    # VARIETY_LIST = ["IF", "TF", "T"]  # 20150320-20150415
    # VARIETY_LIST = ["IF", "TF", "T", "IC", "IH"]  # 20150416-20180816
    # VARIETY_LIST = ["IF", "TF", "T", "IC", "IH", "TS"]  # 20180817- 至今
    if "20100416" <= current_date < "20130906":
        variety_list = ["IF"]
    elif "20130906" <= current_date < "20150320":
        variety_list = ["IF", "TF"]
    elif "20150320" <= current_date < "20150416":
        variety_list = ["IF", "TF", "T"]
    elif "20150416" <= current_date < "20180816":
        variety_list = ["IF", "TF", "T", "IC", "IH"]
    elif "20180816" <= current_date < '20220825':
        variety_list = ["IF", "TF", "T", "IC", "IH", "TS"]
    elif "20220825" <= current_date < '20230421':
        variety_list = ["IF", "TF", "T", "IC", "IH", "IM", "TS"]
    else:
        variety_list = ["IF", "TF", "T", "IC", "IH", "IM", "TS", "TL"]
    return variety_list


class DailyQuoteSpider(object):
    def __init__(self, date: datetime.datetime):
        self.quote_date = datetime.datetime.strptime(date.strftime('%Y-%m-%d'), '%Y-%m-%d')
        self.quote_date_string = self.quote_date.strftime('%Y%m%d')
        self.headers = {'User-Agent': USER_AGENT,
                        'Host': 'www.cffex.com.cn'}

        self.quote_url = "http://www.cffex.com.cn/sj/hqsj/rtj/{}/{}/{}_1.csv".format(
            self.quote_date.strftime('%Y%m'), self.quote_date.strftime('%d'), self.quote_date.strftime('%Y%m%d'))
        self.quote_file = pathlib.Path("cffe/daily/{}.csv".format(self.quote_date_string))

        self.rank_url = "http://www.cffex.com.cn/sj/ccpm/{}/{}/{}_1.csv"
        self.rank_file = 'cffe/rank/{}_{}.csv'  # pathlib.Path("dce/rank/{}.zip".format(self.quote_date_string))

    # 获取日行情文件
    def get_quote_file(self):
        r = requests.post(self.quote_url, headers=self.headers)
        with open(self.quote_file, 'wb') as fp:
            fp.write(r.content)
        print('获取{}.CFE日行情数据完成！'.format(self.quote_date_string))

    # 获取日排名文件
    def get_rank_file(self):
        for v in get_variety_list(self.quote_date_string):
            rank_url = self.rank_url.format(self.quote_date.strftime('%Y%m'), self.quote_date.strftime('%d'), v)
            r = requests.get(rank_url, headers=self.headers)

            rank_file = pathlib.Path(self.rank_file.format(self.quote_date_string, v))
            with open(rank_file, 'wb') as fp:
                fp.write(r.content)
            print('获取{}-{}.CFE日排名数据完成！'.format(self.quote_date_string, v))
            time.sleep(1)


class DailyQuoteParser(object):
    # SERVER_API = "http://127.0.0.1:8000/api/"
    SERVER_API = "https://210.13.218.130:9000/api/"

    def __init__(self, date: datetime.datetime):
        self.quote_date = datetime.datetime.strptime(date.strftime('%Y-%m-%d'), '%Y-%m-%d')
        self.quote_date_string = self.quote_date.strftime('%Y%m%d')

        self.quote_file = pathlib.Path("cffe/daily/{}.csv".format(self.quote_date_string))
        self.rank_file = 'cffe/rank/{}_{}.csv'

        self.quote_save_url = self.SERVER_API + "exchange/cffex/daily/?date={}".format(self.quote_date.strftime("%Y-%m-%d"))
        self.rank_save_url = self.SERVER_API + "exchange/cffex/rank/?date={}".format(self.quote_date.strftime("%Y-%m-%d"))

        self.resolution_quote_file = pathlib.Path("resolution/{}/CFE_Quote.json".format(self.quote_date_string))
        self.resolution_rank_file = pathlib.Path("resolution/{}/CFE_Rank.json".format(self.quote_date_string))
        resolution_folder = pathlib.Path("resolution/{}/".format(self.quote_date_string))
        if not resolution_folder.exists():
            resolution_folder.mkdir(parents=True)

    # 解析日行情文件
    def parse_quote_file(self):
        if not self.quote_file.exists():
            print("没有发现{}.CFE的日交易文件,请先抓取数据!".format(self.quote_date_string))
        csv_df = pd.read_csv(self.quote_file, encoding='gbk', error_bad_lines=False)
        if csv_df.columns.values.tolist() != ['合约代码', '今开盘', '最高价', '最低价', '成交量',
                                              '成交金额', '持仓量', '持仓变化', '今收盘', '今结算', '前结算', '涨跌1', '涨跌2', 'Delta']:
            print("{}.CFE日行情文件格式有误!".format(self.quote_date_string))
            return []
        csv_df = csv_df[~csv_df['合约代码'].str.contains('-C-|-P-|小计|合计')]  # 去除合约代码-C- -P-的期权数据 小计 合计总计数据
        csv_df["合约代码"] = csv_df["合约代码"].str.strip()  # 去除前后空格
        csv_df["成交金额"] = csv_df["成交金额"].round(decimals=6)  # 成交金额保留6位小数
        csv_df["品种"] = csv_df["合约代码"].apply(split_number_en).apply(lambda x: x[0].upper())  # 使用合约代码列添加品种列
        int_date = int(self.quote_date.timestamp())
        csv_df["日期"] = [int_date for _ in range(csv_df.shape[0])]  # 增加日期列
        # 重置索引
        csv_df = csv_df.reindex(columns=["日期", "品种", "合约代码", "前结算", "今开盘", "最高价", "最低价", "今收盘",
                                         "今结算", "涨跌1", "涨跌2", "成交量", "成交金额", "持仓量", "持仓变化"])
        csv_df.columns = ["date", "variety_en", "contract", "pre_settlement", "open_price", "highest", "lowest",
                          "close_price",
                          "settlement", "zd_1", "zd_2", "trade_volume", "trade_price", "empty_volume",
                          "increase_volume"]
        # 填充空值
        csv_df[
            ["pre_settlement", "open_price", "highest", "lowest", "close_price", "settlement", "zd_1", "zd_2",
             "trade_volume", "trade_price", "empty_volume", "increase_volume"]
        ] = csv_df[
            ["pre_settlement", "open_price", "highest", "lowest", "close_price", "settlement", "zd_1", "zd_2",
             "trade_volume", "trade_price", "empty_volume", "increase_volume"]
        ].fillna(0)
        quote_data = csv_df.to_dict(orient='records')
        print('----- 解析{}.CFE行情文件成功,数量:{} -----'.format(self.quote_date_string, len(quote_data)))
        if len(quote_data) > 1:
            # 保存到本地文件
            with open(self.resolution_quote_file, 'w', encoding='utf8') as fp:
                json.dump(quote_data, fp, indent=2, ensure_ascii=False)

    # 解析排名数据
    def parse_rank_file(self):
        result_df = pd.DataFrame()
        for variety_en in get_variety_list(self.quote_date_string):
            rank_file = pathlib.Path(self.rank_file.format(self.quote_date_string, variety_en))
            if not rank_file.exists():
                print('{}_{}.CFE排名文件不存在,请先爬取!')
                raise ValueError("Rank File Not Found!")
            # 解析文件
            variety_df = self._parser_variety_rank_file(rank_file, variety_en)
            result_df = pd.concat([result_df, variety_df])
        if not result_df.empty:
            # 转换数据类型
            result_df["rank"] = result_df["rank"].astype("int")
            result_df["trade"] = result_df["trade"].astype("int")
            result_df["trade_increase"] = result_df["trade_increase"].astype("int")
            result_df["long_position"] = result_df["long_position"].astype("int")
            result_df["long_position_increase"] = result_df["long_position_increase"].astype("int")
            result_df["short_position"] = result_df["short_position"].astype("int")
            result_df["short_position_increase"] = result_df["short_position_increase"].astype("int")
            # 时间戳的日期
            result_df['date'] = result_df['date'].apply(lambda x: int(datetime.datetime.strptime(x, '%Y%m%d').timestamp()))
        rank_data = result_df.to_dict(orient='records')
        print('----- 解析{}.CFE排名文件成功,数量:{} -----'.format(self.quote_date_string, len(rank_data)))
        if len(rank_data) > 1:
            # 保存到本地文件
            with open(self.resolution_rank_file, 'w', encoding='utf8') as fp:
                json.dump(rank_data, fp, indent=2, ensure_ascii=False)

    # 解析一个品种的排名文件
    @staticmethod
    def _parser_variety_rank_file(file_path, variety_name):
        """ 使用pandas解析中金所品种的日排名数据 """
        # 读取数据
        print(file_path)
        variety_df = pd.read_csv(file_path, encoding="gbk", header=None, sep="\t", thousands=',')
        # 得到数据开始的行
        # 搜集数据的容器
        variety_ranks = []
        equal_list = ['交易日', '合约', '排名', '成交量排名', '', '', '持买单量排名', '', '', '持卖单量排名', '', '']
        start_get, first_enter = False, False
        for row in variety_df.itertuples():
            row_list = row[1].split(',')
            if row_list == equal_list:
                start_get = True
                first_enter = True
                continue
            if start_get and first_enter:
                # 修改row_list的值
                row_list[0:3] = equal_list[0:3]  # '交易日', '合约', '排名'
                row_list[3] = row_list[3] + "1"
                row_list[6] = row_list[6] + "2"
                row_list[9] = row_list[9] + "3"
                first_enter = False
                variety_ranks.append(row_list)
                continue
            if start_get and not first_enter:
                variety_ranks.append(row_list)

        variety_df = pd.DataFrame(variety_ranks[1:], columns=variety_ranks[0])
        df_columns = ['交易日', '合约', '排名', '会员简称1', '成交量', '比上一交易日增减', '会员简称2', '持买单量', '比上一交易日增减', '会员简称3', '持卖单量',
                      '比上一交易日增减']
        if variety_df.columns.tolist() != df_columns:
            print("解析中金所{}的日排名数据文件时格式有误!".format(variety_name))
            raise ValueError("Data Columns Error!")
        # 处理数据
        # 1 修改表头
        variety_df.columns = ["date", "contract", "rank", "trade_company", "trade", "trade_increase",
                              "long_position_company", "long_position", "long_position_increase",
                              "short_position_company", "short_position", "short_position_increase"]
        # 2 去除数据空格
        variety_df = variety_df.replace('\s+', '', regex=True)
        # 3 插入品种列
        variety_df["variety_en"] = [variety_name for _ in range(variety_df.shape[0])]
        # 4 重置索引列
        variety_df = variety_df.reindex(columns=["date", "variety_en", "contract", "rank",
                                                 "trade_company", "trade", "trade_increase",
                                                 "long_position_company", "long_position", "long_position_increase",
                                                 "short_position_company", "short_position", "short_position_increase"])

        return variety_df

    # 读取解析好的行情数据
    def read_daily_quote(self):
        if not self.resolution_quote_file.exists():
            raise ValueError('{}.CFE行情数据文件不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_quote_file, 'r', encoding='utf8') as fp:
            quote_data = json.load(fp)
        return quote_data

    # 读取解析好的排名数据
    def read_daily_rank(self):
        if not self.resolution_rank_file.exists():
            raise ValueError('{}.CFE排名数据文件不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_rank_file, 'r', encoding='utf8') as fp:
            rank_data = json.load(fp)
        return rank_data

    # 将行情数据保存到服务器
    def save_daily_quote(self):
        quote_data = self.read_daily_quote()
        if len(quote_data) < 1:
            print('没有发现{}.CFE可以保存的行情数据!'.format(self.quote_date_string))
            return
        try:
            r = requests.post(self.quote_save_url, json=quote_data, verify=False)
            print(r.json())
        except Exception as e:
            print('保存{}.CFE行情数据失败了:{}'.format(self.quote_date_string, e))
        time.sleep(1)

    # 将排名数据保存到服务器
    def save_daily_rank(self):
        rank_data = self.read_daily_rank()
        try:
            r = requests.post(self.rank_save_url, json=rank_data, verify=False)
            print(r.json())
        except Exception as e:
            print('保存{}.CFE排名数据失败了:{}'.format(self.quote_date_string, e))
        time.sleep(1)

    # 总表接口:保存行情数据，所有交易所数据放一个表的
    def new_save_quote(self):
        quote_data = self.read_daily_quote()
        df = pd.DataFrame(quote_data)
        df['quotes_date'] = df['date'].apply(lambda x: datetime.datetime.fromtimestamp(x).strftime('%Y-%m-%d'))
        df.rename(columns={'empty_volume': 'position_volume'}, inplace=True)

        save_url = self.SERVER_API + 'dat/quotes/daily-quotes/'
        try:
            r = requests.post(save_url, json=df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.CFE行情数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.CFE行情数据成功:{},message:{}'.format(self.quote_date_string, r_data['count'], r_data['message']))
        time.sleep(1)

    # 总表接口:保存排名数据，所有交易所数据放一个表的
    def new_save_rank(self):
        rank_data = self.read_daily_rank()
        df = pd.DataFrame(rank_data)
        df['rank_date'] = df['date'].apply(lambda x: datetime.datetime.fromtimestamp(x).strftime('%Y-%m-%d'))
        save_url = self.SERVER_API + 'dat/rank/daily-rank/'
        try:
            r = requests.post(save_url, json=df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.CFE排名数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.CFE排名数据成功:{},message:{}'.format(self.quote_date_string, r_data['count'], r_data['message']))
        time.sleep(1)

    # 总表处理：净持仓数据，所有交易所数据放一个表的
    def run_positions_handler(self):
        # 日行情
        quote_data = self.read_daily_quote()
        quotes_df = pd.DataFrame(quote_data)
        quotes_df.rename(columns={'date': 'quotes_ts', 'empty_volume': 'position_volume'}, inplace=True)
        # 日排名
        rank_data = self.read_daily_rank()
        rank_df = pd.DataFrame(rank_data)
        rank_df.rename(columns={'date': 'rank_ts'}, inplace=True)
        if quotes_df.empty or rank_df.empty:
            print('{}.CFE行情或持仓数据为空!'.format(self.quote_date_string))
            return
        quotes_df = quotes_df[['quotes_ts', 'variety_en', 'contract', 'close_price', 'trade_volume', 'position_volume']]
        rank_df = rank_df[['rank_ts', 'variety_en', 'contract', 'long_position', 'short_position']]
        dominant_df = get_dominant_price_position(quotes_df.copy(), rank_df.copy())
        # 获取合约的持仓数据
        contract_df = get_contract_price_position(quotes_df.copy(), rank_df.copy())
        final_df = pd.concat([dominant_df, contract_df])
        for col in ['close_price', 'position_price', 'position_volume', 'long_position', 'short_position']:
            final_df[col] = final_df[col].apply(lambda x: int(x) if int(x) == float(x) else round(x, 4))
        final_df.sort_values(by='contract', inplace=True)
        # 保存到服务器
        save_url = self.SERVER_API + 'dsas/price-position/'
        try:
            r = requests.post(save_url, json=final_df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.CFE持仓价格数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.CFE持仓价格数据成功:{}'.format(self.quote_date_string, r_data['message']))


if __name__ == '__main__':
    SPIDER = 0
    PARSER = 0
    delta_days = 0
    handle_date = datetime.datetime.today() + datetime.timedelta(days=delta_days)
    if SPIDER:
        spider = DailyQuoteSpider(date=handle_date)
        spider.get_quote_file()  # 获取日行情数据保存为文件
        spider.get_rank_file()   # 获取日排名数据保存为文件
    else:
        parser = DailyQuoteParser(date=handle_date)
        if PARSER:
            parser.parse_quote_file()  # 解析日行情文件数据
            parser.parse_rank_file()  # 解析日排名文件数据
        else:
            # --- 分库分表 ---
            parser.save_daily_quote()  # 保存日行情
            parser.save_daily_rank()  # 保存日排名
            # --- 总表 ---微信机器人GUI.py
            parser.new_save_quote()    # 保存解析的行情数据
            parser.new_save_rank()       # 保存解析的持仓数据
            # -- 净持仓 ---
            parser.run_positions_handler()  # 处理保存净持仓数据
